package com.zyx.pulsardemo.basic.base;

import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;

import java.util.Date;

/**
 * @author Yaxi.Zhang
 * @since 2021/9/30 15:09
 * reference: https://github.com/david-streamlio/pulsar-in-action
 */
public class PulsarDemoBase {

    private final String serviceUrl;
    protected String topic;
    protected String subscriptionName;
    protected String readerName;

    private PulsarClient client;
    private Producer<byte[]> producer;

    public PulsarDemoBase() {
        this.serviceUrl = "pulsar://bigdata022:6650,bigdata023:6650,bigdata024:6650";
        this.topic = "persistent://manning/chapter03/example-topic";
        this.subscriptionName = "my-shared-sub";
        this.readerName = "my-reader";
    }

    public PulsarDemoBase(String serviceUrl, String topic, String subName, String readerName) {
        this.serviceUrl = serviceUrl;
        this.topic = topic;
        this.subscriptionName = subName;
        this.readerName = readerName;
    }

    protected void startProducer() {
        Runnable run = () -> {
            int counter = 0;
            while (true) {
                try {
                    String info = String.format("{id: %d, time: %tc}", ++counter, new Date());
                    System.out.println("send message =====>>>>> " + info);
                    getProducer()
                            .newMessage()
                            .value(info.getBytes())
                            .send();
                    Thread.sleep(10);
                } catch (final Exception ex) {
                    ex.printStackTrace();
                }
            }
        };
        new Thread(run).start();
    }

    protected PulsarClient getClient() throws PulsarClientException {
        if (client == null) {
            synchronized(this) {
                client = PulsarClient.builder().serviceUrl(serviceUrl).build();
            }
        }
        return client;
    }

    protected Producer<byte[]> getProducer() throws PulsarClientException {
        if (producer == null) {
            synchronized(this) {
                producer = getClient().newProducer().topic(topic).create();
            }
        }
        return producer;
    }

}
